--每小时推送数据内容
--1、历史15天的有打标的订单历史状态  去和 订单表匹配新状态  不一样 ==》状态不一致的运单
--2、标签有变的订单直接匹配订单表 ==》标签变化的订单
with order_mark_tmp as (
select
    order_id,waybill_id,concat(label_str,' ',nvl(lag_str,'')) as label_str,dt,order_status_code,pick_network_code
from (
         select
             order_id,waybill_id,label_str,dt,order_status_code,pick_network_code,
             lag(label_str,1,null) over(partition by order_id,waybill_id order by dt) as lag_str,
             row_number() over(partition by order_id,waybill_id order by dt desc) as rk
         from
             (   --历史15天的订单打标结果
                 select order_id ,waybill_id,label_str,dt,order_status_code,pick_network_code
                 from jms_dm.dm_oms_order_mark_detail_hi    ---历史数据
                 where
                     dt between concat(date_add('{{ execution_date | cst_ds }}',-14),'-00')
                            and '{{ execution_date | hour_add(-1) | cst_hour }}'
                 union all
                 --上个小时有新增的订单打标明细
                 select
                       a.order_id
                      ,a.waybill_id as waybill_id
                      ,concat_ws(' ',collect_set(a.field_name))  label_str
                      ,a.dt
                      ,null as order_status_code
                      ,null as pick_network_code
                 from jms_dwd.dwd_yl_oms_order_mark_expand_base_hi a
                 inner join jms_dim.dim_yl_oms_order_mark_dict_base_hi dict
                 on dict.sync_big_data=1 and dict.is_delete = 0 and dict.mark_key=a.field_name
                 where a.dt between '{{ execution_date | cst_hour }}' and'{{ execution_date | cst_hour }}'
                 group by a.order_id,a.waybill_id,a.dt
             )t
     )tt where dt = '{{ execution_date | cst_hour }}' or rk=1
)


insert overwrite table jms_dm.dm_oms_order_mark_detail_hi
select
      t1.order_id
     ,t1.waybill_id
     ,t1.label_str
     ,t2.order_source_code
     ,t2.order_source_name
     ,t2.input_time
     ,t2.order_status_code
     ,case when  t2.order_status_code='100' then '未调派'
           when  t2.order_status_code='106' then '己调派代理区'
           when  t2.order_status_code='101' then '已调派网点'
           when  t2.order_status_code='102' then '已调派业务员'
           when  t2.order_status_code='103' then '已取件'
           when  t2.order_status_code='104' then '已取消'
           when  t2.order_status_code='105' then '取件失败'
      end as order_status_name
     ,t2.pick_network_code
     ,net.name as pick_network_name
     ,t2.sender_province_id
     ,t2.sender_province_name
     ,t2.sender_city_id
     ,t2.sender_city_name
     ,t2.sender_area_id
     ,t2.sender_area_name
     ,t2.sender_detailed_address
     ,t2.receiver_province_id
     ,t2.receiver_province_name
     ,t2.receiver_city_id
     ,t2.receiver_city_name
     ,t2.receiver_area_id
     ,t2.receiver_area_name
     ,t2.receiver_detailed_address
     ,case when t1.order_status_code!=t2.order_status_code or t1.pick_network_code!=t2.pick_network_code then 1
           when t1.dt = '{{ execution_date | cst_hour }}'  then 2
      end as change_type
     ,t2.customer_code
     ,t2.customer_name
     ,net.agent_code as pick_agent_code
     ,net.agent_name as pick_agent_name
     ,'{{ execution_date | cst_hour }}' as dt
from 
order_mark_tmp t1 
left join 
(
    select
    *
    from (
    select  id,waybill_id,
    order_source_code,
    order_source_name,
    input_time,
    order_status_code, --
    pick_network_code, ---
    sender_province_id,
    sender_province_name,
    sender_city_id,
    sender_city_name,
    sender_area_id,
    sender_area_name,
    sender_detailed_address,
    receiver_province_id,
    receiver_province_name,
    receiver_city_id,
    receiver_city_name,
    receiver_area_id,
    receiver_area_name,
    receiver_detailed_address,
    customer_code,
    customer_name,
    row_number() over(partition by id order by dt desc) as rk
    from jms_dwd.dwd_yl_oms_oms_order_hf --小时订单表更新最近15天的数据
    where SUBSTR(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-30) and '{{ execution_date | cst_ds }}'
    )all_way where rk=1
)t2 on t2.id=t1.order_id
left join jms_dim.dim_network_whole_massage  net on net.code= t2.pick_network_code
where t1.dt = '{{ execution_date | cst_hour }}' or t1.order_status_code!=t2.order_status_code or t1.pick_network_code!=t2.pick_network_code
distribute by pmod(hash(order_id), 32)
;
